package org.big.kafkafinancesimulator;

import lombok.extern.slf4j.Slf4j;
import org.big.kafkafinancesimulator.model.MarketData;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Service
public class MarketDataSimulatorService {

    private static final String TOPIC_MARKET_DATA = "market-data";
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final Map<String, MarketData> stockMap = new ConcurrentHashMap<>();

    public MarketDataSimulatorService(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        initializeStocks();
    }

    @Scheduled(fixedRate = 500) // 每5秒更新一次价格
    public void generateMarketData() {
        log.info("Generating market data...   @Scheduled(fixedRate = 500) // 每5秒更新一次价格\n" +
                "    public void generateMarketData() {");
        Random random = new Random();
        stockMap.values().forEach(stock -> {
            // 模拟价格波动，例如在±2%范围内
            double changePercent = (random.nextDouble() - 0.5) * 0.04;
            double newPrice = stock.getCurrentPrice() * (1 + changePercent);
            newPrice = Math.round(newPrice * 100.0) / 100.0; // 保留两位小数

            stock.setCurrentPrice(newPrice);
            stock.setTimestamp(System.currentTimeMillis());

            String stockSymbol = stock.getSymbol();
            // 将更新的行情发送到Kafka
            kafkaTemplate.send(TOPIC_MARKET_DATA,
                    stockSymbol, stock);
        });
    }

    private void initializeStocks() {
        // 初始化几只样本股票
        String[][] initialStocks = {
                {"AAPL", "Apple Inc."},
                {"GOOGL", "Alphabet Inc."},
                {"MSFT", "Microsoft Corporation"}
        };
        double basePrice = 150.0;
        for (String[] stockInfo : initialStocks) {
            stockMap.put(stockInfo[0],
                    new MarketData(stockInfo[0],
                            stockInfo[1],
                            basePrice,
                            System.currentTimeMillis()));
            basePrice += 50.0; // 让每只股票基础价格不同
        }
    }

    public MarketData getCurrentMarketData(String symbol) {
        return stockMap.get(symbol);
    }
}